/*
 * servent.c
 *
 *  Created on: May 20, 2012
 *      Author: burek
 */

#include "cerv/messaging.h"
#include "portutil/sockets.h"
#include "portutil/threading.h"
#include "misc/iniconfig.h"
#include "cerv/message-codes.h"
#include "cerv/node.h"
#include "cerv/location.h"
#include "servent.h"
#include "cerv/logger.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <math.h>

// Polja jednog serventa, radi instanciranja ne smije biti globalno
typedef struct
{
	int svRunning;
	char bootstrapAddress[16];
	int bootstrapPort;
	char myLocation[128];
	char revLocation[128];
	int myPort;
	int locID;
	int mastersPerLevel;
	int master;
	int maxDataValidity;
	int pushInterval;
	int logAll;
	unsigned int tick;
	unsigned int myExternalIP;
	CUSocket *svSock;
	CLocation *root;
	CLocation *myLoc;
	CLocation *myTopLoc;
	CUMutex *treeMut;
	CUMutex *tlogMut;

	float baseTemperature;
	float amplitude;
	float phase;
	float increment;

	int levelMax;
	int levelMin;

	int log;
	int killed;
} SVParams;

// Wraperi za void*
typedef struct
{
	SVParams *sp;
	CUSocket *client;
} SVHandlerWrap;

typedef struct
{
	SVParams *sp;
	char revLocation[512];
} SVAnnounceWrap;

// Zamijeni char u wide-char stringu
int replaceChar(char * text, char from, char to)
{
	int count = 0;
	while(*text)
	{
		if(*text == from)
			*text = to;
		text += 2;
		count++;
	}
	return count;
}

// Wide puts
void wputs(int id, const char * txt)
{
	char buffer[512];
	int i = 0;

	buffer[i++] = '"';
	while(*txt)
	{
		buffer[i++] = *txt;
		txt += 2;
	}
	buffer[i++] = '"';
	buffer[i++] = '\0';
	lputs(id, buffer);
}

// Logovanje odlazne poruke
void cv_logSnd(CUMutex *m, int log, const char * wmsg, int len, const char * addr, int port)
{
	char narrow[256];
	int i;
	cu_lockMutex(m);
	if(len > 500) len = 500;
	for(i = 0; i < len / 2; ++i)
		narrow[i] = wmsg[i * 2];
	narrow[i] = '\0';
	if(CU_OK != cu_checkIPaddress(addr))
			lprintf(log, "Invalid address! \"%s\"", addr);
	lprintf(log, "[%15s : %5d] << \"%s\"", addr, port, narrow);
	cu_unlockMutex(m);
}

// Logovanje dolazne poruke
void cv_logRcv(CUMutex *m, int log, const char * wmsg, int len, const char * addr, int port)
{
	char narrow[256];
	int i;
	cu_lockMutex(m);
	if(len > 500) len = 500;
	for(i = 0; i < len / 2; ++i)
		narrow[i] = wmsg[i * 2];
	narrow[i] = '\0';
	if(CU_OK != cu_checkIPaddress(addr))
		lprintf(log, "Invalid address! \"%s\"", addr);
	lprintf(log, "[%15s : %5d] >> \"%s\"", addr, port, narrow);
	cu_unlockMutex(m);
}

// Provjeri gdje se sve mogu sracunati prosjecne temperature
void cv_updateLevels(SVParams * sp)
{
	CLocation * loc = sp->myLoc->parent;
	CLocation * child = NULL;
	int iter = 0;

	while(loc)
	{
		unsigned int sum = 0;
		unsigned int age = 0;
		unsigned int count = 0;

		iter++;
		if(iter > 50)
		{
			lprintf(sp->log, "Dafuq? Over 50 iterations?!");
			return;
		}

		if(loc->level < sp->levelMax) break;	// Presao nivo ovog serventa

		child = loc->firstChild;
		while(child)
		{
			unsigned int cage = sp->tick - child->timestamp;
			if(cage > sp->maxDataValidity || child->kelvins == 0)	// Ako nisu svi podaci validni
			{														// ne moze ni prosjek biti
				count = 0;
				break;
			}

			if(cage > age) age = cage;
			sum += child->kelvins;
			count++;

			child = child->next;
		}

		if(count > 0)	// Ako jesu validni, izracunaj prosjek i stavi mu starost najstarijeg
		{
			loc->kelvins = sum / count;
			loc->timestamp = sp->tick - age;
			loc = loc->parent;
		}
		else
		{
			break;
		}
	}
}

// Handler za "nestandardne" tekstualne poruke, isto kao u bootstrapu
void cv_altHandler(SVParams * sp, const char * buffer, int length)
{
	char text[1024];
	char *param = NULL;
	int i;

	for(i = 0; i < length / 2; ++i)
		text[i] = buffer[i * 2];
	text[length / 2] = '\0';
	length /= 2;

	param = strchr(text, ' ');
	if(param)
	{
		*param = '\0';
		param++;
	}

	if(strcmp("dumpTree", text) == 0)
	{
		cl_dumpTree(sp->log, sp->root);
	}
	else if(strcmp("logTraffic", text) == 0)
	{
		if(param != NULL)
			sp->logAll = atoi(param);
		else
			lprintf(sp->log, "logTraffic = %d", sp->logAll);
	}
	else if(strcmp("exit", text) == 0)	// Odjavi se i izadji
	{
		sp->svRunning = 0;
	}
	else if(strcmp("die", text) == 0)	// Umri bez odjave
	{
		sp->killed = 1;
		sp->svRunning = 0;
	}
	else if(strcmp("pushInterval", text) == 0)	// Interval slanja svoje temperature
	{
		if(param != NULL)
			sp->pushInterval = atoi(param);
		else
			lprintf(sp->log, "pushInterval = %d sec(s)", sp->logAll);
	}
	else if(strcmp("dataValidity", text) == 0)
	{
		if(param != NULL)
			sp->maxDataValidity = atoi(param);
		else
			lprintf(sp->log, "dataValidity = %d sec(s)", sp->logAll);
	}
	else if(strcmp("relay", text) == 0)		// Prenos poruke, kao u bootstrapu
	{
		if(param != NULL)
		{
			char *port = strchr(param, ':');
			if(port != NULL)
			{
				*port = '\0';
				port++;
				if(CU_OK == cu_checkIPaddress(param))
				{
					char *msgBegin = strchr(port, ' ');
					if(msgBegin != NULL)
					{
						int rport = 0;
						*msgBegin = '\0';
						msgBegin++;

						rport = atoi(port);
						if(rport > 0)
						{
							CUSocket *sock = cu_createTCPSocket();
							if(CU_OK == cu_openClientSocket(sock, param, rport))
							{
								char wideBuff[512];
								int len = strlen(msgBegin);
								int i;
								if(len > 511) len = 511;
								for(i = 0; i < len; ++i)
								{
									wideBuff[i * 2] = msgBegin[i];
									wideBuff[i * 2 + 1] = '\0';
								}

								cu_sendData(sock, wideBuff, len * 2);
								lprintf(sp->log, "Relay successful to %s:%d", param, rport);

								len = cu_recvSpin(sock, wideBuff, 512, 20, 20);
								if(len > 0)
								{
									lprintf(sp->log, "Relay - got response:");
									wputs(sp->log, wideBuff);
								}
								else
								{
									lprintf(sp->log, "Relay - no response");
								}
							}
							else
							{
								lprintf(sp->log, "Relay - error connectiong to %s:%d", param, rport);
							}
							cu_closeSocket(sock);
							free(sock);
						}
						else
						{
							lprintf(sp->log, "Relay - invalid port \"%s\"", port);
						}
					}
					else
					{
						lprintf(sp->log, "Relay - no message specified");
					}
				}
				else
				{
					lprintf(sp->log, "Relay - invalid IP address: \"%s\"", param);
				}
			}
			else
			{
				lprintf(sp->log, "Relay - invalid IP address: \"%s\"", param);
			}
		}
		else
		{
			lprintf(sp->log, "Relay - no parameters specified, format: relay address:port message");
		}
	}
	else
	{
		lprintf(sp->log, "Unrecognized non-standard message - \"%s\"", text);
	}
}

// Upit za temperaturu sa neke udaljene lokacije
int cv_remotePull(SVParams * sp, CLocation * target, const char * revPath)
{
	int kelvins = 0;
	if(target->numNodes > 0)
	{
		int i;

		for(i = 0; i < target->numNodes && kelvins == 0; ++i)	// Kroz sve nodove na lokaciji
		{														// dok ne dobijemo temperaturu
			CUSocket *sock = cu_createTCPSocket();
			if(cu_openClientSocket(sock,
					target->node[i]->addrText, target->node[i]->port) == CU_OK)
			{
				CMessage *pmsg = cm_createMessage(CMD_PULL_DATA);
				char recv[512];
				int reclen = 0;

				cm_appendString(pmsg, revPath);
				cm_finalizeMessage(pmsg);

				// Sastavi zahtjev i posalji ga
				cu_sendData(sock, pmsg->buffer, pmsg->length);
				if(sp->logAll) cv_logSnd(sp->tlogMut, sp->log, pmsg->buffer, pmsg->length, sock->ipTxt, sock->port);
				cm_destroyMessage(pmsg);

				reclen = cu_recvSpin(sock, recv, 512, 20, 20);
				if(sp->logAll) cv_logRcv(sp->tlogMut, sp->log, recv, reclen, sock->ipTxt, sock->port);
				pmsg = cm_decodeMessage(recv, reclen);

				// Odgovor stigao i ispravan?
				if(pmsg != NULL)
				{
					if(pmsg->code == CMD_REPLY_OK)
					{
						// Ne postuje se moja specifikacija...
						if(pmsg->pType[0] == CM_INT)
							kelvins = pmsg->pVal[0].ival;
						else
							kelvins = atoi(pmsg->pVal[0].sval);
						lprintf(sp->log, "Got PULL response from %s (%s:%d)", target->name, target->node[i]->addrText, target->node[i]->port);
					}
					else
					{
						lprintf(sp->log, "Error PULLing from %s (%s:%d)", target->name, target->node[i]->addrText, target->node[i]->port);
					}
				}
				else
				{
					lprintf(sp->log, "Malformed PULL response from %s (%s:%d)", target->name, target->node[i]->addrText, target->node[i]->port);
				}
			}

			cu_closeSocket(sock);
			free(sock);
		}
	}

	return kelvins;
}

// Ko je na vrhu podstabla?
CLocation *cv_getTopLevel(CLocation *loc)
{
	if(loc->parent == NULL) return NULL;

	while(loc->parent->parent != NULL)
		loc = loc->parent;
	return loc;
}

// Koji je sljedeci node na putu od 'from' do 'to' kroz stablo?
CLocation * cv_getNextConnection(int id, CLocation * from, CLocation * to)
{
	CLocation *tl1 = NULL, *tl2 = NULL;
	lprintf(id, "Connection from %s to %s?", from->name, to->name);

	tl1 = cv_getTopLevel(from);
	tl2 = cv_getTopLevel(to);

	if(tl1 != tl2)
	{
		if(from == tl1)
			return tl2;
		else
			return from->parent;
	}

	if(cl_isAncestorOf(from, to))
	{
		while(to && to->parent != from)
			to = to->parent;

		lprintf(id, "Down the tree %s -> %s", from->name, to->name);
		return to;
	}
	else
	{
		if(from->parent != NULL)
		{
			if(from->parent->parent == NULL)
			{
				CLocation *child = from->parent->firstChild;
				while(child)
				{
					CLocation *tto = to;
					while(tto && tto->parent != child)
						tto = tto->parent;
					if(tto != NULL)
					{
						lprintf(id, "Sideways %s -> %s", from->name, child->name);
						return child;
					}
				}
				lprintf(id, "Cannot find the next link!");
				return NULL;
			}
			else
			{
				lprintf(id, "Up the tree %s -> %s", from->name, from->parent->name);
				return from->parent;
			}
		}
		else
		{
			lprintf(id, "Parent is NULL? Noone else to ask");
			return NULL;
		}
	}
}

// Obavi PULL, moze biti kesiran ili se mora traziti preko mreze
int cv_pullData(SVParams * sp, const char * path)
{
	CLocation *target = cl_createLocation(sp->root, path);
	char revPath[512];
	cl_reverseAddress(path, revPath);

	lprintf(sp->log, "pullData() for %s", path);

	if(target == sp->myLoc)		// Pull se odnosi na lokaciju ovog serventa, to vec znamo mjerenjem
	{
		lprintf(sp->log, "Direct pull from : %s", target->name);
		return target->kelvins;
	}
	// Imamo dovoljno svjezu informaciju u cacheu?
	else if(target->kelvins > 0 && sp->tick - target->timestamp <= sp->maxDataValidity)
	{
		lprintf(sp->log, "Cached pull from : %s (for %s)", sp->myLoc->name, revPath);
		return target->kelvins;
	}
	else
	{
		// Ako znamo adresu nekoga na toj lokaciji, pitacemo direktno
		if(target->numNodes > 0)
		{
			int kelvins = cv_remotePull(sp, target, revPath);

			if(kelvins > 0)
			{
				return kelvins;
			}
			else
			{
				lprintf(sp->log, "Couldn't PULL from any know nodes");
				return 0;
			}
		}
		else
		{
			// Ako ne, gledamo ko bi sljedeci mogao znati, pa pitamo njega
			CLocation *loc = cv_getNextConnection(sp->log, sp->myTopLoc, target);
			lprintf(sp->log, "Node unknown, should consult others, topLoc = %s", sp->myTopLoc->name);
			if(loc == NULL)
			{
				lprintf(sp->log, "Cannot find a connection to \"%s\"", path);
				return 0;
			}
			else
			{
				int result = 0;

				lprintf(sp->log, "Consulting %s...", loc->name);
				result = cv_remotePull(sp, loc, revPath);
				if(result > 0)
				{
					lprintf(sp->log, "Successfuly consulted %s, got result", loc->name);
				}
				else
				{
					lprintf(sp->log, "Failed to consult %s", loc->name);
				}
				return result;
			}
		}
	}

	return 0;
}

// Opsluzivanje zahtjeva
void * cv_requestHandler(void * params)
{
	SVHandlerWrap *svh = (SVHandlerWrap*)params;
	CUSocket *sock = svh->client;
	CMessage *msg, *reply = NULL;
	CLocation *loc = NULL;
	SVParams *sp = svh->sp;

	char buffer[4096];
	int len = 0, port;

	len = cu_recvData(sock, buffer, 4096);
	if(sp->logAll) cv_logRcv(sp->tlogMut, sp->log, buffer, len, sock->ipTxt, sock->port);

	// Ako je tekstualna, ide u altHandler
	if(len > 0)
	{
		if(buffer[0] < '0' || buffer[0] > '9')
		{
			cv_altHandler(sp, buffer, len);
			return NULL;
		}
	}

	msg = cm_decodeMessage(buffer, len);

	if(msg == NULL)
	{
		lprintf(sp->log, "SV) Malformed message received, ignoring");
	}
	else
	{
		switch(msg->code)
		{
		case CMD_PING:
			{
				// Na ping odmah odgovori pongom
				reply = cm_createMessage(CMD_PONG);
				cm_appendString(reply, msg->pVal[0].sval);
				cm_finalizeMessage(reply);
				cu_sendData(sock, reply->buffer, reply->length);
				if(sp->logAll) cv_logSnd(sp->tlogMut, sp->log, reply->buffer, reply->length, sock->ipTxt, sock->port);
			}
			break;

		case CMD_REGISTER_SERVENT:
			{
				// Neki drugi servent se prijavljuje
				char rev[1024];
				CMessage *msg2;
				int len;

				// Rastavljanje poruke na parametre (lokacija:master:port)
				cu_lockMutex(svh->sp->treeMut);
				memcpy(rev, buffer, msg->length * 2);
				len = replaceChar(rev, ':', ';');
				msg2 = cm_decodeMessage(rev, len * 2);

				cl_reverseAddress(msg2->pVal[0].sval, rev);
				loc = cl_createLocation(svh->sp->root, rev);

				if(msg2->pType[2] == CM_INT)
					port = msg2->pVal[2].ival;
				else
					port = atoi(msg2->pVal[2].sval);

				{
					CNode *nnode = cl_findNode(loc, sock->ipAddr, port);
					if(nnode == NULL)
					{
						if(loc->numNodes >= svh->sp->mastersPerLevel)	// Vec imamo dovoljno mastera na tom nivou?
						{												// Ne bi se trebalo desiti ako je sve ispravno
							lprintf(sp->log, "SV %d) Location full, cannot register @%s (%s:%d)",
									svh->sp->locID, msg2->pVal[0].sval,
									loc->node[0]->addrText, loc->node[0]->port);
						}
						else
						{
							// Ako ima mjesta, dodajemo ga u datu lokaciju
							char buff[512];
							sprintf(buff, "%u:%d:%s", sock->ipAddr, port, msg2->pVal[0].sval);
							nnode = cn_constructNode(buff);
							nnode->role = msg2->pVal[1].sval[0] == '1';
							cl_attachNode(loc, nnode, svh->sp->mastersPerLevel);

							lprintf(sp->log, "SV %d) P2P register OK %s:%d @ %s",
									svh->sp->locID, nnode->addrText, nnode->port, msg2->pVal[0].sval);
						}
					}
				}
				cm_destroyMessage(msg2);

				cu_unlockMutex(svh->sp->treeMut);
				fflush(stdout);
			}
			break;

		case CMD_PUSH_DATA:
			{
				// Stizu podaci o temperaturama od nekog drugog serventa
				int i;
				for(i = 0; i < msg->paramCount; ++i)
				{
					char rev[512];
					char *split = NULL;
					CLocation *loc = NULL;
					if(msg->pType[i] != CM_STRING) continue;

					split = strchr(msg->pVal[i].sval, ':');
					*split = '\0';

					cl_reverseAddress(msg->pVal[i].sval, rev);
					loc = cl_getByAddr(sp->root, rev);	// Trazimo lokaciju na koju se odnosi

					if(loc != NULL)
					{
						if(sscanf(split + 1, "%d", &(loc->kelvins)))
						{
							// Ako je temperatura ispravna, upisujemo je i lupamo trenutni timestamp
							loc->timestamp = sp->tick;
							cv_updateLevels(sp);	// Mozda sada mozemo sracunati novi prosjek?
						}
					}
				}
			}
			break;

		case CMD_PULL_DATA:
			{
				// Neko salje upit za temperaturu neke lokacije
				char revAddress[512];
				int result = 0;
				CMessage *reply = NULL;
				if(msg->pType[0] == CM_STRING)
				{
					cl_reverseAddress(msg->pVal[0].sval, revAddress);
					result = cv_pullData(sp, revAddress);	// Pozivamo funkciju za to, iznad

					if(result > 0)
					{
						reply = cm_createMessage(CMD_REPLY_OK);
						cm_appendInt(reply, result);
						cm_finalizeMessage(reply);
						cu_sendData(sock, reply->buffer, reply->length);
						if(sp->logAll) cv_logSnd(sp->tlogMut, sp->log, reply->buffer, reply->length, sock->ipTxt, sock->port);
						lprintf(sp->log, "PULL successful, reply (%d / %.1f C) sent", result,
								(float)result / 100.0f - 273.15f);
					}
					else
					{
						reply = cm_createMessage(CMD_REPLY_FAIL);
						cm_appendString(reply, "0");
						cm_finalizeMessage(reply);
						cu_sendData(sock, reply->buffer, reply->length);
						if(sp->logAll) cv_logSnd(sp->tlogMut, sp->log, reply->buffer, reply->length, sock->ipTxt, sock->port);
						lprintf(sp->log, "PULL failed, reply sent");
					}
				}
				else
				{
					lprintf(sp->log, "PULL fail - invalid or no location requested");
				}

				cm_destroyMessage(reply);
			}
			break;

		case CMD_POKE:
			{
				// Marijin omiljeni feature
				lprintf(sp->log, "SV) Got poked by %s:%d", sock->ipTxt, sock->port);
				fflush(stdout);
			}
			break;

		case CMD_REPLY_OK:
			break;

		default:
			{
				// Nepoznat tip poruke? Ispisati, debuggovanja radi i optimisticno odgovoriti OK
				lprintf(sp->log, "SV %d) Got the following message from %s:%d :",
						svh->sp->locID, sock->ipTxt, sock->port); fflush(stdout);
				cm_dumpParams(sp->log, msg);

				reply = cm_createMessage(CMD_REPLY_OK);
				cm_finalizeMessage(reply);
				cu_sendData(sock, reply->buffer, reply->length);
				if(sp->logAll) cv_logSnd(sp->tlogMut, sp->log, reply->buffer, reply->length, sock->ipTxt, sock->port);
			}
			break;
		}
	}

	cu_closeSocket(sock);
	free(sock);
	cm_destroyMessage(reply);

	free(svh);
	return NULL;
}

// Loop za slusanje zahtjeva i spawnovanje opsluzivaca
void * cv_socketListener(void * params)
{
	SVParams *sp = (SVParams*)params;
	CUSocket *sock = sp->svSock, *client = NULL;

	while(sp->svRunning)
	{
		SVHandlerWrap *svh = (SVHandlerWrap*)calloc(1, sizeof(SVHandlerWrap));
		client = cu_acceptConnection(sock);
		svh->sp = sp;
		svh->client = client;
		cu_spawnThread(cv_requestHandler, (void*)svh);
	}

	return NULL;
}

// Provjeri koji je moj gornji level u stablu
void cv_checkLevel(SVParams * sp)
{
	CLocation *loc = sp->myLoc;
	sp->levelMax = sp->levelMin = sp->myLoc->level;
	sp->myTopLoc = sp->myLoc;

	while(loc->parent)
	{
		if(!cl_findNode(loc->parent, sp->myExternalIP, sp->myPort))
			break;
		sp->myTopLoc = loc->parent;
		sp->levelMax--;
		loc = loc->parent;
	}
}

// Parsuj odgovor bootstrapa sa spiskom master servenata
void cv_registerFromMessage(SVParams *sp, CMessage * reply)
{
	int i;
	CLocation *testLoc;

	// Za svaku pojedinacnu stavku...
	for(i = 0; i < reply->paramCount; ++i)
	{
		char address[512];
		CNode *n = NULL, *tmpn = NULL;
		n = cn_constructNode(reply->pVal[i].sval);	// Napravi node
		CLocation *l = cl_createLocation(sp->root, n->location); // Napravi/nadji lokaciju
		tmpn = cl_recFindNode(sp->root, n->addrInt, n->port);	// Da li je taj node poznat od prije?
		if(tmpn != NULL)
		{
			//cn_destroyNode(n);
			n = tmpn;
		}

		n->role = 1;

		if(!cl_findNode(l, n->addrInt, n->port))	// Ako ga nema vec, prikaci, jer duple prijave su moguce
			cl_attachNode(l, n, 99);

		cl_constructFullAddress(l, address);

		if(strcmp(address, sp->myLocation) == 0)	// Ako sam to ja, sada znam i svoju eksternu adresu
		{
			sp->myExternalIP = n->addrInt;
			sp->master = 1;
		}
	}

	testLoc = cl_createLocation(sp->root, sp->revLocation);		// Moja lokacija u svijetu
	sp->myLoc = testLoc;

	cv_checkLevel(sp);
}

// Oglasi se ostalim serventima, radi se nakon uspjesne registracije kod bootstrapa
void cv_announceToServents(CLocation * loc, void * vparams)
{
	SVAnnounceWrap *aw = (SVAnnounceWrap*)vparams;
	SVParams * sp = aw->sp;
	CUSocket * sock;
	CMessage * msg;
	int i;
	char tmp[512];
	char address[512];

	if(loc == sp->myLoc) return;

	cl_reverseAddress(aw->revLocation, address);

	if(loc->numNodes > 0)
	{
		sprintf(tmp, "%s:%d:%d", address, sp->master, sp->myPort);
		msg = cm_createMessage(CMD_REGISTER_SERVENT);
		cm_appendString(msg, tmp);
		cm_finalizeMessage(msg);
	}

	// Za sve nodove na datoj lokaciji
	for(i = 0; i < loc->numNodes; ++i)
	{
		// Ako su masteri (inace im se nema potrebe javljati) i ako to nije ovaj isti node, javi se
		if(loc->node[i]->role > 0 && (loc->node[i]->addrInt != sp->myExternalIP || loc->node[i]->port != sp->myPort))
		{
			sock = cu_createTCPSocket();
			if(cu_openClientSocket(sock, loc->node[i]->addrText, loc->node[i]->port) == CU_OK)
			{
				cu_sendData(sock, msg->buffer, msg->length);
				if(sp->logAll) cv_logSnd(sp->tlogMut, sp->log, msg->buffer, msg->length, sock->ipTxt, sock->port);
			}
			else
			{
				lprintf(sp->log, "SV %d) Cannot announce to %s:%d @ %s", sp->locID,
						loc->node[i]->addrText, loc->node[i]->port,
						loc->name);
			}
			cu_closeSocket(sock);
			free(sock);
		}
	}

	if(loc->numNodes > 0)
	{
		cm_destroyMessage(msg);
	}

}

// Obidji cijelo stablo i pozovi announce
void cv_propagateAnnouncment(SVParams * sp, CLocation * loc)
{
	SVAnnounceWrap aw;

	cu_lockMutex(sp->treeMut);

	aw.sp = sp;
	while(loc->parent)
	{
		CLocation *child;
		char address[512];

		cl_constructFullAddress(loc, address);
		cl_reverseAddress(address, aw.revLocation);

		cv_announceToServents(loc->parent, &aw);

		child = loc->parent->firstChild;
		while(child)
		{
			if(child != sp->myLoc)
			{
				cv_announceToServents(child, &aw);
			}
			child = child->next;
		}

		if(!cl_findNode(loc->parent, sp->myExternalIP, sp->myPort))
			break;

		loc = loc->parent;
	}

	cu_unlockMutex(sp->treeMut);
}

// Registruj se kod bootstrapa, odmah po pokretanju
int cv_register(SVParams * sp)
{
	CUSocket *sock = NULL;
	CMessage *msg, *reply;
	char buffer[4096];
	int len = 0;

	// Konektuj se
	sock = cu_createTCPSocket();
	if(cu_openClientSocket(sock, sp->bootstrapAddress, sp->bootstrapPort) == CU_ERROR)
	{
		lprintf(sp->log, "SV) Failed to connect to bootstrap"); fflush(stdout);
		return 0;
	}

	// Sastavi poruku za registraciju
	msg = cm_createMessage(CMD_BOOTSTRAP_REGISTER_SERVENT);
	sprintf(buffer, "%d:%s", sp->myPort, sp->myLocation);
	cm_appendString(msg, buffer);

	cm_finalizeMessage(msg);
	cu_sendData(sock, msg->buffer, msg->length);
	if(sp->logAll) cv_logSnd(sp->tlogMut, sp->log, msg->buffer, msg->length, sock->ipTxt, sock->port);
	cm_destroyMessage(msg);

	// Primi odgovor
	len = cu_recvData(sock, buffer, 4096);
	if(sp->logAll) cv_logRcv(sp->tlogMut, sp->log, buffer, len, sock->ipTxt, sock->port);
	reply = cm_decodeMessage(buffer, len);

	if(reply != NULL)
	{
		if(reply->code == CMD_REPLY_OK)
		{
			CLocation *loc;
			lprintf(sp->log, "SV %d) Got OK, registered", sp->locID);

			cu_lockMutex(sp->treeMut);

			// Ako je OK, to znaci da uz poruku ide spisak ostalih servenata, pa ga procesuiraj
			cv_registerFromMessage(sp, reply);
			cu_closeSocket(sock);
			free(sock);
			cm_destroyMessage(reply);

			// Ako nisam medju dojavljenim serventima, pravim lokaciju za sebe
			if(sp->myLoc == NULL)
			{
				CLocation *mloc;
				mloc = cl_createLocation(sp->root, sp->revLocation);
				sp->myLoc = mloc;
			}
			else if(sp->myLoc->numNodes == 0)
			{
				char repLine[512];
				CNode *mnd;
				sprintf(repLine, "%d:%d:%s", 0, sp->myPort, sp->revLocation);
				mnd = cn_constructNode(repLine);
				mnd->role = 0;
				sp->master = 0;
				if(!cl_attachNode(sp->myLoc, mnd, 1))
					lprintf(sp->log, "ATTACH FAILED!");
			}

			loc = sp->myLoc;

			if(loc == NULL)
				lprintf(sp->log, "I HAVE NO LOCATION! :O");

			cu_unlockMutex(sp->treeMut);

			cu_sleep(sp->locID * 47);
			cv_propagateAnnouncment(sp, loc);	// Javi se ostalim serventima

			return 1;
		}
		else
		{
			// Ako registracija nije uspjela, dalji rad nema svrhe
			lprintf(sp->log, "SV %d) Got negative response, NOT registered, ERROR %d", sp->locID, reply->pVal[0].ival);
			cu_closeSocket(sock);
			free(sock);
			cm_destroyMessage(reply);
			return 0;
		}
	}

	return 0;
}

// Odjava kod bootstrapa
int cv_unregister(SVParams * sp)
{
	CUSocket *sock = NULL;
	CMessage *msg, *reply;
	char buffer[4096] = {0};
	int len = 0;

	// Konektuj se
	sock = cu_createTCPSocket();
	if(cu_openClientSocket(sock, sp->bootstrapAddress, sp->bootstrapPort) == CU_ERROR)
	{
		lprintf(sp->log, "SV) Failed to connect to bootstrap");
		return 0;
	}

	// Sastavi poruku
	msg = cm_createMessage(CMD_BOOTSTRAP_UNREGISTER_SERVENT);
	cm_appendString(msg, sp->myLocation);
	cm_finalizeMessage(msg);
	cu_sendData(sock, msg->buffer, msg->length);
	if(sp->logAll) cv_logSnd(sp->tlogMut, sp->log, msg->buffer, msg->length, sock->ipTxt, sock->port);
	cm_destroyMessage(msg);

	lprintf(sp->log, "SV) Goodbye sent");

	len = cu_recvData(sock, buffer, 4096);
	if(sp->logAll) cv_logRcv(sp->tlogMut, sp->log, buffer, len, sock->ipTxt, sock->port);
	reply = cm_decodeMessage(buffer, len);

	// Odgovor?
	if(reply != NULL)
	{
		if(reply->code == CMD_REPLY_OK)
		{
			cu_closeSocket(sock);
			free(sock);
			cm_destroyMessage(reply);
			return 1;
		}
		else
		{
			// Sta god...
			lprintf(sp->log, "SV %d) Got negative response, NOT unregistered? ERROR %d", sp->locID, reply->pVal[0].ival);
			cu_closeSocket(sock);
			free(sock);
			cm_destroyMessage(reply);
			return 0;
		}
	}

	cu_closeSocket(sock);
	free(sock);
	return 0;
}

// Posalji svoju temperaturu svim ostalim serventima na svim nivoima
// na kojima je ovaj servent master + jedan iznad
void cv_pushMyData(SVParams * sp)
{
	CLocation *loc;
	CNode *mnode = cl_findNode(sp->myLoc, sp->myExternalIP, sp->myPort);
	CMessage *msg = NULL;
	char tmp[512];
	char num[10];
	int i;

	loc = sp->myLoc;

	while(loc)
	{
		if(sp->tick - loc->timestamp > sp->maxDataValidity || loc->kelvins == 0) break;
		if(loc->level < sp->levelMax) break;

		// Sastavi poruku
		msg = cm_createMessage(CMD_PUSH_DATA);
		cl_constructFullAddress(loc, tmp);
		strcat(tmp, ":");
		sprintf(num, "%05d", loc->kelvins);
		strcat(tmp, num);
		cm_appendString(msg, tmp);
		cm_finalizeMessage(msg);

		// Javi svim nodovima na lokaciji
		for(i = 0; i < loc->parent->numNodes; ++i)
		{
			if(loc->parent->node[i] != mnode)
			{
				CUSocket *sock = cu_createTCPSocket();
				if(cu_openClientSocket(sock,
						loc->parent->node[i]->addrText,
						loc->parent->node[i]->port) == CU_OK)
				{
					cu_sendData(sock, msg->buffer, msg->length);
					if(sp->logAll) cv_logSnd(sp->tlogMut, sp->log, msg->buffer, msg->length, sock->ipTxt, sock->port);
				}
				cu_closeSocket(sock);
				free(sock);
			}
		}

		cm_destroyMessage(msg);
		loc = loc->parent;		// Idi prema gore
	}
}

// Entry point za servent
void * serventInstanceEntry(void * params)
{
	int seconds;
	CUSocket *svSock;
	CUThread *listenerThread = NULL;
	SVLaunchParams *lparams;
	SVParams sp;
	ININode *ini;

	if(params == NULL) return NULL;
	ini = ini_parseFile("cervent.ini");		// Ucitaj konfiguraciju

	lparams = (SVLaunchParams*)params;
	svSock = cu_createTCPSocket();

	seconds = ini_getInt(ini, "cervent/serventExpire", -1);

	// Parametri, defaulti, iz INIja...
	sp.myPort = lparams->port;
	strcpy(sp.bootstrapAddress, ini_getString(ini, "bootstrap/address", "127.0.0.1"));
	sp.bootstrapPort = ini_getInt(ini, "bootstrap/port", 9666);
	sp.svSock = svSock;
	strcpy(sp.myLocation, lparams->location);
	cl_reverseAddress(sp.myLocation, sp.revLocation);
	sp.svRunning = 1;
	sp.locID = lparams->locID;
	sp.root = cl_createRoot();
	sp.myLoc = NULL;
	sp.mastersPerLevel = 3;
	sp.master = 0;
	sp.tick = 0;
	sp.myExternalIP = 0;
	sp.treeMut = cu_createMutex();
	sp.tlogMut = cu_createMutex();

	sp.logAll = ini_getInt(ini, "logging/log_traffic", 0);
	sp.pushInterval = ini_getInt(ini, "servent/push_interval", 3);
	sp.maxDataValidity = ini_getInt(ini, "servent/data_validity", 10);

	// "Simulacija" temperature
	sp.amplitude = lparams->amplitude;
	sp.baseTemperature = lparams->baseTemperature;
	sp.phase = lparams->phase;
	sp.increment = lparams->increment;

	sp.log = l_register();
	sp.killed = 0;

	// Ako nema socketa, nista
	if(!svSock)
		{ lprintf(sp.log, "Cannot start servent"); l_disconnect(sp.log); return NULL; }
	if(cu_openServerSocket(svSock, sp.myPort) == CU_ERROR)
		{ lprintf(sp.log, "Cannot start servent"); l_disconnect(sp.log); return NULL; }

	cu_sleep(1000);

	// Ako registracija nije uspjela, prekidaj thread, nema onda sta da se radi
	if(!cv_register(&sp)) { l_disconnect(sp.log); return NULL; }

	// Pokreni listener
	listenerThread = cu_spawnThread(cv_socketListener, &sp);

	sp.svRunning = 1;

	lprintf(sp.log, "Servent #%d started for location %s", sp.locID, sp.myLocation);
	l_setTitle(sp.log, sp.myLoc->name);
	l_registerPort(sp.log, sp.myPort);

	lprintf(sp.log, "My level span: %d -> %d", sp.levelMin, sp.levelMax);
	lprintf(sp.log, "Push interval: %d sec(s)", sp.pushInterval);

	while(sp.svRunning)
	{
		sp.tick++;

		if(seconds > 0)		// Obsolete
		{
			seconds--;
			if(seconds <= 0)
				sp.svRunning = 0;
		}

		// Oscilovanje temperature
		sp.phase += sp.increment;
		if(sp.phase > 6.28318531f)
			sp.phase -= 6.28318531f;
		sp.myLoc->kelvins = (int)((sin(sp.phase) * sp.amplitude + sp.baseTemperature) * 100.0f);
		sp.myLoc->timestamp = sp.tick;
		cv_updateLevels(&sp);

		// Ako je vrijeme za slanje svoje temperature
		if(sp.tick % sp.pushInterval == 0)
			cv_pushMyData(&sp);

		if(sp.tick == 5 + sp.locID)	// Debugovanja radi, ispisi svoje stablo par sekundi nakon pokretanja
		{
			lprintf(sp.log, "===============< %2d [%d] %s %u:%d >===============", sp.locID, sp.master, sp.myLocation, sp.myExternalIP, sp.myPort);
			cl_dumpTree(sp.log, sp.root);
			lprintf(sp.log, "===========================================================================");
		}

		cu_sleep(1000);
	}
	sp.svRunning = 0;
	if(!sp.killed) cv_unregister(&sp);	// Ako je naredba bila 'die', nemoj se odjaviti

	l_disconnect(sp.log);

	cu_killThread(listenerThread);

	return NULL;
}

